package detector

import (
    //"google.golang.org/grpc"
    "sync"
    "../pb"
    "../common"
    "golang.org/x/net/context"
    "time"
)

type FailureDetectorService struct {
    records                map[common.Node]*Record
    mu                     sync.RWMutex
    myNode                 common.Node
    isActiveMaster         bool
    onConnectedCallback    func(common.Node)
    onDisconnectedCallback func([]common.Node)
    opt                    Option
    stopping               chan interface{}
    stopped                chan interface{}
}

func (fds *FailureDetectorService) ack(msg *pb.BeaconMsg) *pb.BeaconAck {
    ack := &pb.BeaconAck{
        Time:     msg.Time,
        ThisNode: msg.ToNode,
    }
    fds.mu.RLock()
    defer fds.mu.RUnlock()
    ack.PrimeNode = string(fds.myNode)
    ack.IsMaster = fds.isActiveMaster
    return ack
}

func (fds *FailureDetectorService) SetActive(activeMode bool) {
    fds.mu.Lock()
    defer fds.mu.Unlock()
    fds.isActiveMaster = activeMode
}

func (fds *FailureDetectorService) checkAll() { //repeatly
    fds.mu.Lock()
    defer fds.mu.Unlock()
    var expiredNodes []common.Node
    for node, rcd := range fds.records {
        if now := common.NowMillisecond(); rcd.isAlive != false && now-rcd.lastTime > fds.opt.GraceMilliseconds {
            rcd.isAlive = false
            expiredNodes = append(expiredNodes, node)
        }
    }
    if len(expiredNodes)>0{
        fds.onDisconnectedCallback(expiredNodes)
    }
}

func (fds *FailureDetectorService) onConnected(node common.Node) {
    fds.mu.Lock()
    defer fds.mu.Unlock()
    rcd, ok := fds.records[node]
    if !ok {
        fds.records[node] = NewRecord()
        fds.onConnectedCallback(node)
        return
    }
    if now := common.NowMillisecond(); now > rcd.lastTime {
        rcd.lastTime = now
        if rcd.isAlive == false {
            rcd.isAlive = true
            fds.onConnectedCallback(node)
        }
    }
}


func (fds *FailureDetectorService) Ping(ctx context.Context, msg *pb.BeaconMsg) (*pb.BeaconAck, error) {
    node := common.Node(msg.FromNode)
    fds.onConnected(node)
    return fds.ack(msg), nil
}

func (fds *FailureDetectorService) Stop() {
    close(fds.stopping)
    <-fds.stopped
}

func NewFailureDetectorService(myNode common.Node, opt Option,
    onConnectedCallback func(common.Node), onDisconnectedCallback func([]common.Node)) *FailureDetectorService {
    fds := &FailureDetectorService{
        records:                make(map[common.Node]*Record),
        myNode:                 myNode,
        opt:                    opt,
        onConnectedCallback:    onConnectedCallback,
        onDisconnectedCallback: onDisconnectedCallback,
        stopping:               make(chan interface{}),
        stopped:                make(chan interface{}),
    }
    return fds
}

func (fds *FailureDetectorService) Start() {
    tk := time.Tick(time.Duration(fds.opt.CheckIntervalMilliseconds) * time.Millisecond)
LOOP:
    for {
        select {
        case <-tk:
            fds.checkAll()
        case <-fds.stopping:
            break LOOP
        }
    }
    close(fds.stopped)
}
